package cn.com.sparkle.paxos.paxosinstance;

import java.io.IOException;
import java.util.LinkedList;
import java.util.Set;
import java.util.concurrent.Future;

import org.apache.log4j.Logger;

import com.google.protobuf.ByteString;
import com.google.protobuf.GeneratedMessage;

import cn.com.sparkle.paxos.AccountBook;
import cn.com.sparkle.paxos.addprocess.AddRequestPackage;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.event.InstancePaxosEvent;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Id;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstancePrepareRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceSuccessMessage;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.InstanceVoteRequest;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.SuccessfulRecord;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.Value;
import cn.com.sparkle.paxos.iomodel.protobuffmessage.PaxosMessages.ValueType;
//import cn.com.sparkle.paxos.model.Id;
//import cn.com.sparkle.paxos.model.Value;
import cn.com.sparkle.paxos.net.NetNode;
//import cn.com.sparkle.paxos.net.message.PrepareRequest;
//import cn.com.sparkle.paxos.net.message.SuccessMessage;
//import cn.com.sparkle.paxos.net.message.VoteRequest;
//import cn.com.sparkle.paxos.net.message.instance.InstancePrepareRequest;
//import cn.com.sparkle.paxos.net.message.instance.InstanceSuccessMessage;
//import cn.com.sparkle.paxos.net.message.instance.InstanceVoteRequest;
//import cn.com.sparkle.paxos.net.protobuffmessage.PaxosMessages.InstanceSuccessMessage;
//import cn.com.sparkle.paxos.net.protobuffmessage.PaxosMessages.Value;
//import cn.com.sparkle.paxos.net.protobuffmessage.PaxosMessages.ValueType;
//import cn.com.sparkle.paxos.stablestorage.model.SuccessfulRecord;

public class InstancePaxosInstance extends PaxosInstance {
	private final static Logger logger = Logger
			.getLogger(InstancePaxosInstance.class);

	private AccountBook aBook;
	private LinkedList<AddRequestPackage> addRequestPackages;

	private EventsManager eventsManager;
	private Value wantedValue;

	private long debugTime = System.currentTimeMillis();

	private Id.Builder id;

	private long startTime;

	public InstancePaxosInstance(Set<NetNode> activeNode, int quorum,
			long instanceId, Id.Builder id, AccountBook aBook,
			LinkedList<AddRequestPackage> addRequestPackages,
			EventsManager eventsManager, String selfAddress) {
		super(activeNode, quorum, instanceId, selfAddress);
		this.id = id;
		this.aBook = aBook;
		this.addRequestPackages = addRequestPackages;
		int count = 0;
		ValueType valueType = ValueType.TYPE_COMM;
		for (AddRequestPackage arp : addRequestPackages) {
			if (arp.isManageCommand()) {
				valueType = ValueType.TYPE_ADMIN;
			}
			count += arp.valueList.size();
		}
		// byte[][] vs = new byte[count][];
		count = 0;
		Value.Builder vb = Value.newBuilder();
		// vb.setType(valueType);
		for (AddRequestPackage arp : addRequestPackages) {
			for (ByteString bb : arp.valueList) {
				// ByteString bs = ByteString.copyFrom(bb);
				vb.addValues(bb);
				// vb.setValues(count++, bb);
				// vs[count++] = bb;
			}
		}
		vb.setType(valueType);
		wantedValue = vb.build();
		// wantedValue = new Value(valueType, vs);
		this.eventsManager = eventsManager;
	}

	@Override
	public GeneratedMessage getSuccessMessage(Value lastValue) {
		InstanceSuccessMessage.Builder b = InstanceSuccessMessage.newBuilder();
		b.setId(instanceId).setValue(lastValue);
		// return new InstanceSuccessMessage(instanceId, lastValue);
		return b.build();
	}

	public LinkedList<AddRequestPackage> getAddRequestPackages() {
		return addRequestPackages;
	}

	@Override
	public GeneratedMessage getPrepareRequest() {
		InstancePrepareRequest request = InstancePrepareRequest.newBuilder()
				.setInstanceId(instanceId).setId(id).build();
		return request;
	}

	@Override
	public GeneratedMessage getVoteRequest(Value lastValue) {
		InstanceVoteRequest.Builder builder = InstanceVoteRequest.newBuilder();
		builder.setInstanceId(instanceId).setVoteId(id).setValue(lastValue);
		// return new InstanceVoteRequest(instanceId, id, lastValue);
		return builder.build();
	}

	@Override
	public Value getWantAssginValue() {
		return wantedValue;
	}

	@Override
	public void voteSuccess(Value value) {
		try {
			SuccessfulRecord successfulRecord = SuccessfulRecord.newBuilder()
					.setV(value).build();

			if (value == wantedValue) {
				aBook.writeSuccessfulRecord(instanceId, successfulRecord,
						addRequestPackages);

				// successfulRecord = new SuccessfulRecord(instanceId, value,
				// addRequestPackages);
			} else {
				// successfulRecord = new SuccessfulRecord(instanceId, value,
				// null);
				aBook.writeSuccessfulRecord(instanceId, successfulRecord, null);
			}

			logger.debug("cost :" + (System.currentTimeMillis() - debugTime));
			InstancePaxosEvent.doSuccessEvent(eventsManager, this, value);
		} catch (IOException e) {
			logger.error("fatal error:", e);
		}
	}

	@Override
	public void instanceFail(long refuseId) {
		InstancePaxosEvent.doFailEvent(eventsManager, this, id, refuseId);
	}

	@Override
	public Future<Boolean> activate() {
		startTime = System.currentTimeMillis();
		InstancePaxosEvent.doStartEvent(eventsManager, this);
		return super.activate();
	}

	public long getStartTime() {
		return startTime;
	}

}
